package hbec.kettle.etl.plugins.reader.mysqlreader;

import java.text.MessageFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import hbec.kettle.etl.common.constants.DataBase;
import hbec.kettle.etl.common.exception.ExceptionTracker;
import hbec.kettle.etl.common.plugin.PluginStatus;
import hbec.kettle.etl.common.plugin.Reader;
import hbec.kettle.etl.engine.conf.ReaderConf;
import hbec.kettle.etl.engine.plugin.Line;
import hbec.kettle.etl.engine.plugin.LineExchanger;
import hbec.kettle.etl.plugins.common.DALUtils;
import hbec.kettle.etl.plugins.common.WatchEntity;

public class MysqlReader extends Reader {
	private static final Logger logger = LoggerFactory.getLogger(MysqlReader.class);
	
	private Map<DataBase, String> watchSumTablesOfDb;
	//private Map<DataBase, String> checkPointTableOfDb;
	
	private Map<String, Long> positionOfObj = new HashMap<>();
	private Map<String, WatchEntity> watches = new ConcurrentHashMap<>();
	
	private Long position = 0L;
	
	public int prepare(ReaderConf readerConf) {
		this.watchSumTablesOfDb = readerConf.getWatchSumTablesOfDb();
		//this.checkPointTableOfDb = readerConf.getCheckPointTableOfDb();
		for (DataBase e : watchSumTablesOfDb.keySet()) {
			int result = DALUtils.initializeSqlite(DataBase.SQLITE, MessageFormat.format("{0}", e.value().toUpperCase()));
			if (result <= 0) 
				logger.warn("Sqlite initialize fail");
		}
		return PluginStatus.SUCCESS.value();
	}
	
	public int init() {
		return PluginStatus.SUCCESS.value();
	}
	
	public void startRead() {
		while (true) {
			try {
				for (Map.Entry<DataBase, String> e : watchSumTablesOfDb.entrySet()) {
					watches = DALUtils.getWatchSum(e.getKey(), e.getValue(), position);
					
					String prefix = e.getKey().value().concat(".");
					for (Map.Entry<String, WatchEntity> we : watches.entrySet()) {
						Long _position = we.getValue().getMaxPosition();
						if (_position.compareTo(position) > 0) 
							position = _position;
						
						Map<Integer, Set<String>> _identities = new HashMap<>();
						if (flush(we.getValue().getObject(), position)) {
							for (Map.Entry<Integer, Set<String>> kv : we.getValue().getIdentities().entrySet()) {
								if (kv.getKey().intValue() <= 3) {
									if (_identities.containsKey(Integer.valueOf(3))) 
										_identities.get(Integer.valueOf(3)).addAll(kv.getValue());
									else {
										Set<String> temp = new HashSet<>();
										temp.addAll(kv.getValue());
										_identities.put(Integer.valueOf(3), temp);
									}
								} else {
									_identities.put(kv.getKey(), kv.getValue());
								}
							}
							
							Line line = new Line();
							line.setId(we.getValue().getObject().concat("-".concat(String.valueOf((new Date()).getTime()))));
							line.setObject(prefix.concat(we.getValue().getObject()));
							line.setMinPosition(we.getValue().getMinPosition());
							line.setMaxPosition(we.getValue().getMaxPosition());
							line.setIdentities(_identities);
							line.setIds(we.getValue().getIds());
							
							while (!LineExchanger.sendToWriter(line)) {
								Thread.sleep(5 * 1000);
							}
						}
					}
				}

				Thread.sleep(2 * 1000);
			} catch (Exception e) {
				logger.error(ExceptionTracker.trace(e));
			}
		}
	}
	
	private boolean flush(String obj, Long position) {
		if (positionOfObj.containsKey(obj)) {
			if (position.compareTo(positionOfObj.get(obj)) > 0) {
				positionOfObj.put(obj, position);
				return true;
			} else 
				return false;
		} else {
			positionOfObj.put(obj, position);
			return true;
		}
	}
}
